﻿using RedLockNet.SERedis;
using StackExchange.Redis;
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using Tsual.DataPipeline;

namespace Tsual.DataPipeline.Redis
{
    public abstract class AbstractCheckedRedisHashPipe<T, TMeta> : AbstractRedisPipe<T, TMeta>
        where TMeta : IConvertible<KeyValuePair<string, string>>, IConvertible<LockType>
    {
        
        public AbstractCheckedRedisHashPipe(IDatabase cache, RedLockFactory lockFactory) : base(cache, lockFactory)
        {
        }

        public override bool CanRead(TMeta meta)
        {
            if (meta == null) return false;
            var pmeta = ((IConvertible<KeyValuePair<string, string>>)meta).Convert();
            return !string.IsNullOrEmpty(pmeta.Key) && !string.IsNullOrEmpty(pmeta.Value);
        }

        public override bool CanWrite(T data, TMeta meta)
        {
            if (meta == null) return false;
            var pmeta = ((IConvertible<KeyValuePair<string, string>>)meta).Convert();
            return !string.IsNullOrEmpty(pmeta.Key) && !string.IsNullOrEmpty(pmeta.Value);
        }
    }

    public class RedisHashPipe<TMeta> : AbstractCheckedRedisHashPipe<string, TMeta>
        where TMeta : IConvertible<KeyValuePair<string, string>>, IConvertible<LockType>
    {
        public RedisHashPipe(IDatabase cache, RedLockFactory lockFactory) : base(cache, lockFactory)
        {
        }

        public override Task<DataPipeReadResult<string>> ReadCache(TMeta meta) => Task.Run(() =>
        {
            var pmeta = ((IConvertible<KeyValuePair<string, string>>)meta).Convert();
            var value = cache.HashGet(pmeta.Key, pmeta.Value);
            if (value.HasValue) return DataPipeReadResult<string>.Success(value);
            else return DataPipeReadResult<string>.Notfound();
        });

        public override Task<DataPipeWriteResult> WriteCache(string data, TMeta meta) => Task.Run(() =>
        {
            var pmeta = ((IConvertible<KeyValuePair<string, string>>)meta).Convert();
            cache.HashSet(pmeta.Key, pmeta.Value, data);
            return DataPipeWriteResult.Success();
        });

        protected override string GetReadLockKey(TMeta meta) => ((IConvertible<KeyValuePair<string, string>>)meta).Convert().ToString();
    }

    /// <summary>
    /// readonly direct write T to redis
    /// </summary>
    /// <typeparam name="T"></typeparam>
    /// <typeparam name="TMeta"></typeparam>
    /// <typeparam name="TConvertor"></typeparam>
    public class RedisExpirableDirectwriteHashPipe<T, TMeta, TConvertor> : AbstractRedisPipe<T, TMeta>
        where TMeta : IConvertible<LockType>
        where TConvertor : class, IConvertor<T, ExpirablePipeValue<string>>, IConvertor<T,KeyValuePair<string, string>>, new()
    {
        TConvertor convertor = new TConvertor();
        IConvertor<T, ExpirablePipeValue<string>> valueConvertor => convertor as IConvertor<T, ExpirablePipeValue<string>>;
        IConvertor<T, KeyValuePair<string, string>> keyConvertor => convertor as IConvertor<T, KeyValuePair<string, string>>;
        public RedisExpirableDirectwriteHashPipe(IDatabase cache, RedLockFactory lockFactory) : base(cache, lockFactory)
        {
        }

        public override bool CanRead(TMeta meta) => false;

        public override bool CanWrite(T data, TMeta meta)
        {
            var key = keyConvertor.Convert(data);
            return !string.IsNullOrEmpty(key.Key) && !string.IsNullOrEmpty(key.Value);
        }

        public override Task<DataPipeReadResult<T>> ReadCache(TMeta meta)
        {
            throw new NotImplementedException();
        }

        public override Task<DataPipeWriteResult> WriteCache(T data, TMeta meta) => Task.Run(() =>
        {
            var key = keyConvertor.Convert(data);
            var value = valueConvertor.Convert(data);
            if (value.AbsoluteExpireTime > DateTimeOffset.Now)
            {
                cache.HashSet(key.Key, key.Value, value.Value);
                if (value.AbsoluteExpireTime.HasValue)
                    cache.HashSet(ExpHead + key.Key, key.Value, value.AbsoluteExpireTime.Value.ToUnixTimeSeconds());
            }
            return DataPipeWriteResult.Success();
        });

        protected override string GetReadLockKey(TMeta meta)
        {
            throw new NotImplementedException();
        }

        protected override string GetWriteLockKey(T data, TMeta meta)
        {
            return keyConvertor.Convert(data).ToString();
        }
    }

    public class RedisExpirableHashPipe<T, TMeta, TConvertor> : AbstractCheckedRedisHashPipe<T, TMeta>
        where TMeta : IConvertible<KeyValuePair<string, string>>, IConvertible<LockType>
        where TConvertor : class, IConvertor<T, ExpirablePipeValue<string>>, new()
    {
        private static TConvertor convertor = new TConvertor();

        public RedisExpirableHashPipe(IDatabase cache, RedLockFactory lockFactory) : base(cache, lockFactory)
        {
        }

        public override Task<DataPipeReadResult<T>> ReadCache(TMeta pmeta) => Task.Run(() =>
        {
            var meta = ((IConvertible<KeyValuePair<string, string>>)pmeta).Convert();
            var value = cache.HashGet(meta.Key, meta.Value);
            var exp = (long?)cache.HashGet(ExpHead + meta.Key, meta.Value);

            if (!exp.HasValue || exp.Value > DateTimeOffset.Now.ToUnixTimeSeconds())
            {
                return value.HasValue ? DataPipeReadResult<T>.Success(convertor.Convert(new ExpirablePipeValue<string>() { Value = value, AbsoluteExpireTime = DateTimeOffset.Now.AddSeconds(exp.Value) })) : DataPipeReadResult<T>.Notfound();
            }
            else
            {
                cache.HashDelete(meta.Key, meta.Value);
                cache.HashDelete(ExpHead + meta.Key, meta.Value);
                return DataPipeReadResult<T>.Notfound();
            }
        });

        public override Task<DataPipeWriteResult> WriteCache(T pdata, TMeta pmeta) => Task.Run(() =>
        {
            var meta = ((IConvertible<KeyValuePair<string, string>>)pmeta).Convert();
            var data = convertor.Convert(pdata);
            if (data.AbsoluteExpireTime > DateTimeOffset.Now)
            {
                cache.HashSet(meta.Key, meta.Value, data.Value);
                if (data.AbsoluteExpireTime.HasValue)
                    cache.HashSet(ExpHead + meta.Key, meta.Value, data.AbsoluteExpireTime.Value.ToUnixTimeSeconds());
            }
            return DataPipeWriteResult.Success();
        });

        protected override string GetReadLockKey(TMeta meta) => ((IConvertible<KeyValuePair<string, string>>)meta).Convert().ToString();
    }

    public class RedisExpirableHashPipe<TMeta> : RedisExpirableHashPipe<ExpirablePipeValue<string>, TMeta, DefaultEqualConvertor<ExpirablePipeValue<string>>>
        where TMeta : IConvertible<KeyValuePair<string, string>>, IConvertible<LockType>
    {
        public RedisExpirableHashPipe(IDatabase cache, RedLockFactory lockFactory) : base(cache, lockFactory)
        {
        }
    }
}
